package com.alison.datastream.etl;

import com.alison.datastream.domain.TaxiRide;
import com.alison.datastream.util.TaxiRideGenerator;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author alison
 * @Date 2024/4/3 23:32
 * @Version 1.0
 * @Description
 */
public class ETLmapApp {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<TaxiRide> rides = env.addSource(new TaxiRideGenerator());

        DataStream<EnrichedRide> enrichedNYCRides = rides.filter(new NYCFilter()).map(new Enrichment());
        enrichedNYCRides.print();
        env.execute("map function demo");
    }

    public static class NYCFilter implements FilterFunction<TaxiRide> {
        @Override
        public boolean filter(TaxiRide taxiRide) throws Exception {
            return true;
        }
    }

    public static class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {

        @Override
        public void flatMap(TaxiRide taxiRide, Collector<EnrichedRide> out) throws Exception {
            FilterFunction<TaxiRide> valid = new NYCFilter();
            if (valid.filter(taxiRide)) {
                out.collect(new EnrichedRide(taxiRide));
            }
        }
    }

    public static class Enrichment implements MapFunction<TaxiRide, EnrichedRide> {

        @Override
        public EnrichedRide map(TaxiRide taxiRide) throws Exception {
            return new EnrichedRide(taxiRide);
        }
    }
}
